#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <functional>
#include <thread>
#include "logger.hpp"

//RabbitMQ客户端API接口的二次封装: 为了更适合项目中的使用(1.也就是我们只用传关键参数就可以到达我们想要的目的 2.变成我们自己熟悉的代码样子)
//为什么能简化操作? 因为我们的项目需要使用的功能很简单, 所以只需要部分参数, 其他参数/其他操作都是固定的, 所以能简化

namespace wufan_im{
// 封装一个MQClient: 提供声明指定交换机与队列, 并进行绑定的功能
// 提供向指定交换机发布消息的功能
// 提供订阅指定队列消息, 并设置回调函数进行消息消费处理的功能
class MQClient{
    public:
        using MessageCallback = std::function<void(const char*, size_t)>;
        MQClient(const std::string& user, const std::string passwd, const std::string host){
            _loop = EV_DEFAULT; //实例化底层网络通信框架I/O事件监控句柄
            _handler = std::make_unique<AMQP::LibEvHandler>(_loop); //实例化libEvHandler句柄 -- 将AMQP框架与事件监控关联起来
            std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";   
            AMQP::Address address(url);
            _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);   //实例化连接对象
            _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());   //实例化信道对象

            //创建一个线程, 执行这个loop, 用来监听服务器  我们之前请求的响应
            _loop_thread = std::thread([this](){
                ev_run(_loop, 0);
            });
        }
        // 退出的时候, 不能直接退出, 因为它退出的时候, 还有一些要处理, 所以我们用异步的退出
        ~MQClient(){
            struct ev_async async_watcher;
            ev_async_init(&async_watcher, watcher_callback);    //初始化async_watcher, 指定当异步事件发生时应该调用的回调函数为watcher_callback
            ev_async_start(_loop, &async_watcher);  //这行代码将async_watcher注册到事件循环_loop中，使其开始监听异步事件
            ev_async_send(_loop, &async_watcher);//这行代码手动触发一个异步事件，通知_loop中的async_watcher，进而调用之前注册的回调函数watcher_callback
            _loop_thread.join();    //阻塞当前线程, 等待_loop_thread线程结束
        }
        //声明组件
        void declareComponents(const std::string& exchange, const std::string& queue, 
            const std::string& routing_key = "routing_key", AMQP::ExchangeType exchange_type = AMQP::ExchangeType::direct) {
            // 声明交换机 - 声明的作用: 服务器若不存在这个test-exchange交换机则会创建, 并且还可以设置一些配置
            auto& exchange_deferred = _channel->declareExchange(exchange, exchange_type); //发送请求, 并不是立即响应, 而是会在ev中监测到, 监测到后会调用回调函数
            //AMQP::ExchangeType::direct表示直连交换机
            exchange_deferred.onError([](const char* message){
                LOG_ERROR( "声明交换机失败: {}\n", message);
                exit(0);
            });
            exchange_deferred.onSuccess([exchange](){   //不能用引用, 因为是回调函数, 这是拷贝指针, 引用可能释放
                LOG_DEBUG( "{} 交换机创建成功! \n", exchange);
            });

            // 声明队列
            auto& queue_deferred = _channel->declareQueue(queue);
            queue_deferred.onError([](const char* message){
                LOG_ERROR( "声明队列失败: {}\n", message);
                exit(0);
            });
            queue_deferred.onSuccess([queue](){
                LOG_DEBUG( "{} 队列创建成功! \n", queue);
            });

            // 针对交换机和队列进行绑定
            // 告诉先找这个 exchange, 再根据 routing_key 则找到对应的queue
            auto& binding_deferred = _channel->bindQueue(exchange, queue, routing_key);
            binding_deferred.onError([exchange, queue](const char* message){
                LOG_ERROR( "{} = {} 绑定失败: {}\n", exchange, queue, message);
                exit(0);
            });
            binding_deferred.onSuccess([exchange, queue](){
                LOG_DEBUG( "{} = {} 绑定成功!\n", exchange, queue );
            });
        }
        //向交换机发布消息
        bool publish(const std::string& exchange, const std::string& msg, const std::string& routing_key = "routing_key") {
            bool ret = _channel->publish(exchange, routing_key, msg);
            if (ret == false) {
                LOG_ERROR("{} 发布消息失败: ", exchange);
                return false;
            }
            return true;
        }
        // 订阅队列消息 -- 设置消息处理回调函数
        //创建一个队列的消费者, consume-tag代表消费者唯一标识
        void consume(const std::string& queue, const MessageCallback& cb) {
            auto& consumer_deferred = _channel->consume("test-queue", "consume-tag");
            consumer_deferred.onError([](const char* message){
                std::cout << "订阅 test-queue 队列消息失败: " << message << std::endl;
                exit(0);
            });
            //收到消息后该如何处理
            consumer_deferred.onReceived([this, cb](const AMQP::Message& message, uint64_t deliveryTag, bool redelivered){
                cb(message.body(), message.bodySize()); //可以看到, 我们实际上不需要使用这么多参数, 我们只需要使用这两个参数就可以了
                _channel->ack(deliveryTag);
            });
        }
    private:
        static void watcher_callback(struct ev_loop* loop, ev_async* watcher, int32_t revents) {
            ev_break(loop, EVBACKEND_ALL);  //这行代码用于停止事件循环
        }
    private:
        struct ev_loop* _loop;
        std::thread _loop_thread;
        std::unique_ptr<AMQP::LibEvHandler> _handler;   //1.不定义为智能指针, 就必须要在初始化列表初始化 2.不支持拷贝
        std::unique_ptr<AMQP::TcpConnection> _connection;
        std::unique_ptr<AMQP::TcpChannel> _channel;
};
}